package main

import (
	zc "ac"
	log "ac-common-go/glog"
	"ac-common-go/version"
	"ac-julink/cache"
	"ac-julink/handlers"

	//"ac-julink/handlers"
	"ac-julink/manager"
	"ac-julink/mq"
	"ac-julink/mqlistener"
	"ac-julink/rpcagent/wgctl"
	"ac-julink/rpcagent/wgdm"
	"ac-julink/rpcagent/wgmcu"
	"ac-julink/rpcagent/wgota"
	"ac-julink/socket"
	config "ac-julink/tool/config"
	config_blacklist "ac-julink/tool/config_blacklist"
	config_http "ac-julink/tool/config_http"
	config_sarama "ac-julink/tool/config_sarama"
	"ac-julink/tool/data"
	"ac-julink/tool/procinfo"
	"ac-julink/tool/util"
	"flag"
	"os"
	"os/signal"
	"runtime"
	"strconv"
	"syscall"
	//"ac-julink/rpcagent/wgmcu"
	"net/http"
	_ "net/http/pprof"

	"ac-julink/kafka"
	svc "ac-julink/service"
	"net"
)

type NetService struct {
	//goroutine同步,等待协程处理结束
	zc.ZService
}

func main() {
	ver := flag.Bool("version", false, "current service build version")
	flag.Parse()
	if *ver {
		version.ShowVersion(false)
		return
	}
	version.ShowVersion(true)
	//3.读取配置信息
	LoadConfig(false)
	proc, b := config.GetConfigIntValue("WGNS", "proc")
	if b != true {
		log.Errorf("get proc failed")
	}

	svc.InitConfigs()

	runtime.GOMAXPROCS(proc * runtime.NumCPU())
	config_http.Init()
	config_blacklist.Init()
	config_sarama.Init()

	//初始化map缓存
	context := cache.NewSyncContext()

	acRestClient := manager.NewAcRestClient()

	// app gateway service start
	netService := NewService(context)
	if netService == nil {
		log.Error("new app service failed")
		return
	}

	//pprof stack info
	go func() {
		err := http.ListenAndServe(netService.GetIP()+":"+"6060", nil)
		if err != nil {
			log.Error("pprof http.listenAndServe failed")
		}
	}()

	//读取ActiveMQ登录用户名和密码
	encpassword, username, password, mqSSL := readMqConfig()

	userName, err := util.CryptoJaspy(encpassword, username)
	if err != nil {
		log.Errorf("Decode ENC username failed,err:%v", err)
	}
	log.Info("encpassword:%v", encpassword)
	log.Info("userName:%v", userName)
	passWord, err := util.CryptoJaspy(encpassword, password)
	if err != nil {
		log.Errorf("Decode ENC password failed,err:%v", err)
	}
	log.Info("passWord:%v", passWord)

	//4.初始化ActiveMQ连接池
	mq.InitMqPool(userName, passWord, mqSSL)

	//5.初始化ActiveMQ Listener
	mq.StartWgdataMqListener(mqlistener.NewWgdataListener(), context, acRestClient, userName, passWord, mqSSL)
	mq.StartWgctlMqListener(mqlistener.NewWgctlListener(), context, acRestClient, userName, passWord, mqSSL)
	mq.StartWgssMqListener(mqlistener.NewWgssListener(), context, acRestClient, userName, passWord, mqSSL)
	mq.StartWgotaMqListener(mqlistener.NewWgotaListener(), context, acRestClient, userName, passWord, mqSSL)
	mq.StartWgCommonMqListener(mqlistener.NewWgcommonListener(), context, acRestClient, userName, passWord, mqSSL)
	mq.StartWgmcuMqListener(mqlistener.NewWgmcuListener(), context, acRestClient, userName, passWord, mqSSL)
	// dispose blacklist config and clean cache
	mq.StartBlackListNCleanCacheListener(mqlistener.NewBlackListNCleanCacheListener(), context, acRestClient, userName, passWord, mqSSL)

	//6.初始化HTTP连接
	wgdm.InitTriftPool()
	wgctl.InitTriftPool()
	wgota.InitTriftPool()
	wgmcu.InitTriftPool()

	kafka.InitProducers()

	//启动每分钟打印REQ STATUS
	go procinfo.Init()
	go reloadConfigNotify()
	go dumpStackNotify()

	//初始化socketManager
	socket.SocketManager = socket.NewAcceptor()

	hbToPsThreshold, _ := config.GetConfigIntValue("WGNS", "hbToPsThreshold")
	if hbToPsThreshold == 0 {
		hbToPsThreshold = 60000
	}

	//8.启动TLS服务,等待连接，端口号443
	go func() {
		//8.启动TLS服务,等待连接
		startTlsServer(":443", "./conf/cert/server.pem", "./conf/cert/server.key", context, acRestClient)
	}()
	//8.启动TLS服务,等待连接，端口号8443
	go func() {
		//8.启动TLS服务,等待连接
		startTlsServer(":8443", "./conf/cert/server.pem", "./conf/cert/server.key", context, acRestClient)
	}()
	//启动http服务注册至ETCD，端口号5813
	netService.Start()
	//主进程等待协程处理结束信号
}

func readMqConfig() (string, string, string, int) {
	encpassword := config.GetConfigValue("WGNS", "mq_enc_password")
	username := config.GetConfigValue("WGNS", "mq_broker_username")
	password := config.GetConfigValue("WGNS", "mq_broker_password")
	mqSSL, b := config.GetConfigIntValue("WGNS", "mqSSL")
	if b != true {
		log.Errorf("mqSSL not found, default to 0")
		mqSSL = 0
	}
	return encpassword, username, password, mqSSL
}

func startTlsServer(addr, pemPath, keyPath string, context *cache.SyncContext, client *manager.AcRestClient) { //启动tls服务,等待连接
	maxSessionNum, tokenBucketCapacity, tokenFillIntervalSec := readConnConfig()
	log.Infof("TLS addr:[%s]", addr)
	socket.SocketManager.StartTls(addr, pemPath, keyPath, context, client, maxSessionNum, int64(tokenBucketCapacity), int64(tokenFillIntervalSec))

}

func readConnConfig() (int, int, int) {
	var maxSessionNum, tokenBucketCapacity, tokenFillIntervalSec int
	var ok bool
	if maxSessionNum, ok = config.GetConfigIntValue("WGNS", "max_session_num"); !ok {
		maxSessionNum = 200000
	}
	if tokenBucketCapacity, ok = config.GetConfigIntValue("WGNS", "tokenBucketCapacity"); !ok {
		tokenBucketCapacity = 1000
	}
	if tokenFillIntervalSec, ok = config.GetConfigIntValue("WGNS", "tokenFillIntervalSec"); !ok {
		tokenFillIntervalSec = 1
	}

	return maxSessionNum, tokenBucketCapacity, tokenFillIntervalSec
}

func NewService(cacheContext *cache.SyncContext) *NetService {
	//common.CheckParam(config != nil)
	service := &NetService{}
	servicePort, _ := config_http.GetConfigValueInt64("ServicePort")
	serviceAddr := config_http.GetConfigValue("ServiceAddr")
	if len(serviceAddr) <= 0 {
		serviceAddr = getMachineIP()
		if len(serviceAddr) <= 0 {
			log.Errorf("get config self service addr failed by machine IP.")
			return nil
		}
	}
	var serverConfig = &zc.ZServiceConfig{IP: serviceAddr,
		Port: strconv.FormatInt(servicePort, 10)}
	service.Init(config_http.GetConfigValue("ServiceName"), serverConfig)

	julinkRestHandler := handlers.NewJulinkRestHandler()

	service.Handle("cleanCaches", zc.ZServiceHandler(func(req *zc.ZMsg, resp *zc.ZMsg) {
		julinkRestHandler.HandleCleanCaches(req, resp, cacheContext)
	}))

	service.Handle("getCachedValueByKey", zc.ZServiceHandler(func(req *zc.ZMsg, resp *zc.ZMsg) {
		julinkRestHandler.HandleGetCachedValueByKey(req, resp, cacheContext)
	}))
	service.Handle("fetchAllCachedKeys", zc.ZServiceHandler(func(req *zc.ZMsg, resp *zc.ZMsg) {
		julinkRestHandler.HandleFetchAllCachedKeys(req, resp, cacheContext)
	}))

	service.Handle("cleanAllCaches", zc.ZServiceHandler(func(req *zc.ZMsg, resp *zc.ZMsg) {
		julinkRestHandler.HandleCleanAllCaches(req, resp, cacheContext)
	}))

	return service
}
func dumpStackNotify() {
	sigs := make(chan os.Signal, 1)
	signal.Notify(sigs, syscall.SIGUSR2)
	buf := make([]byte, 1<<24)
	for {
		<-sigs
		stacklen := runtime.Stack(buf, true)
		log.Errorf("=== received SIGUSR2 ===\n*** goroutine dump...\n%s\n*** end\n", buf[:stacklen])
	}
}

func reloadConfigNotify() {
	sigs := make(chan os.Signal, 1)
	signal.Notify(sigs, syscall.SIGUSR1)
	for {
		<-sigs
		log.Infof("=== received SIGUSR1 === *** will reload config ***")
		LoadConfig(true)
	}
}

func LoadConfig(isReload bool) error { //读配置信息
	//2.初始化配置文件信息
	config.InitConfigInfo("./conf/wgns_config.conf")
	data.Mysid = config.GetConfigValue("WGNS", "sid")
	statusCtlFlag, ok := config.GetConfigIntValue("WGNS", "statustoctlflag")
	if !ok {
		statusCtlFlag = 1
		log.Error("Get statustoctlflag failed!")
	}
	data.StatusToCtlFlag = statusCtlFlag
	//增加tls版本开关
	tlsTurn, ok := config.GetConfigIntValue("WGNS", "tlsturn")
	if !ok {
		tlsTurn = 0
		log.Error("Get tls turn failed!")
	}
	data.TlsTurn = tlsTurn
	writeTimeout, ok := config.GetConfigIntValue("WGNS", "tcp_write_timeout")
	if !ok {
		writeTimeout = -1
		log.Error("Get tcp_write_timeout failed!")
	}
	data.WriteTimeout = writeTimeout

	sendMsgTimeout, ok := config.GetConfigIntValue("WGNS", "tcp_send_timeout")
	if !ok {
		sendMsgTimeout = 1
		log.Error("Get tcp_send_timeout failed!")
	}
	data.SendMsgTimeout = sendMsgTimeout
	//901kafka topic
	statusTopic := config.GetConfigValue("WGNS", "dshd_status_topic")
	data.KafkaTopic901 = statusTopic
	//201kafka topic
	cmdResultTopic := config.GetConfigValue("WGNS", "dshd_cmd_result_topic")
	data.KafkaTopic201 = cmdResultTopic
	//903kafka topic
	statusAddTopic := config.GetConfigValue("WGNS", "dshd_status_903_topic")
	data.KafkaTopic903 = statusAddTopic
	socket.InitSocketOptions()
	log.Info("config load success")
	return nil
}

func getMachineIP() string {
	netInterfaces, err := net.Interfaces()
	if err != nil {
		log.Errorf("net.Interfaces failed, err:[%v]", err.Error())
		return ""
	}
	for i := 0; i < len(netInterfaces); i++ {
		if (netInterfaces[i].Flags & net.FlagUp) != 0 {
			addrs, _ := netInterfaces[i].Addrs()
			for _, address := range addrs {
				if ipnet, ok := address.(*net.IPNet); ok && !ipnet.IP.IsLoopback() {
					if ipnet.IP.To4() != nil {
						log.Infof("the machine IP is: [%s]", ipnet.IP.String())
						return ipnet.IP.String()
					}
				}
			}
		}
	}
	return ""
}
